#ifndef _XBASE_TABLET_SCANNER_HPP_
#define _XBASE_TABLET_SCANNER_HPP_

#include <list>

#include <boost/function.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/system/error_code.hpp>
#include <boost/asio/deadline_timer.hpp>

#include "stat.hpp"
#include "protocol.hpp"
#include "scan_struct.hpp"
#include "client_options.hpp"
#include "session_manager.hpp"
#include "io_service_pool.hpp"
#include "async_module_base.hpp"

namespace xbase
{
	// 非线程安全，只允许单线程使用，并且只有在一个请求完成以后才能够进行下一次异步请求
	class tablet_scanner
		: public async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level> base_type;
		typedef tablet_scanner this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef session_manager::endpoint_type endpoint_type;

		typedef protocol::packet_type<protocol::SCAN_OPEN_REQ>::type scan_open_request_type;
		typedef protocol::packet_type<protocol::SCAN_OPEN_REQ>::ptr_type scan_open_request_ptr;
		typedef protocol::packet_type<protocol::SCAN_OPEN_RESP>::type scan_open_resp_type;
		typedef protocol::packet_type<protocol::SCAN_OPEN_RESP>::ptr_type scan_open_resp_ptr;

		typedef protocol::packet_type<protocol::SCAN_NEXT_REQ>::type scan_next_request_type;
		typedef protocol::packet_type<protocol::SCAN_NEXT_REQ>::ptr_type scan_next_request_ptr;
		typedef protocol::packet_type<protocol::SCAN_NEXT_RESP>::type scan_next_resp_type;
		typedef protocol::packet_type<protocol::SCAN_NEXT_RESP>::ptr_type scan_next_resp_ptr;

		typedef protocol::packet_type<protocol::SCAN_CLOSE_REQ>::type scan_close_request_type;
		typedef protocol::packet_type<protocol::SCAN_CLOSE_REQ>::ptr_type scan_close_request_ptr;
		typedef protocol::packet_type<protocol::SCAN_CLOSE_RESP>::type scan_close_resp_type;
		typedef protocol::packet_type<protocol::SCAN_CLOSE_RESP>::ptr_type scan_close_resp_ptr;

		typedef boost::system::error_code error_code;

		typedef boost::function<void(row_result::ptr_type, const error_code&)> handler_type;
		typedef std::list<handler_type> handler_list;

	public:
		tablet_scanner(io_service_pool &pool,
			const endpoint_type &ep,
			session_manager &sm,
			tablet_scan_info &si)
			: base_type(pool),
			m_server_endpoint(ep),
			m_session_manager(sm),
			m_options(pool.options()),
			m_scan_info(si),
			m_returned_row_count(0),
			m_started(false),
			m_retry_times(0)
		{
			m_scan_open_pkt_ptr = scan_open_request_ptr(new scan_open_request_type);
			m_row_result_ptr = row_result::ptr_type(new row_result);

			// 初始化scan_open数据包
			protocol::scan_open_request &scan_open_pkt = m_scan_open_pkt_ptr->body();
			scan_open_pkt.m_tablet_name = m_scan_info.m_tablet_name;
			scan_open_pkt.m_start_row = m_scan_info.m_start_row;
			scan_open_pkt.m_end_row = m_scan_info.m_end_row;
			scan_open_pkt.m_column_names = m_scan_info.m_column_names;
			scan_open_pkt.m_latest_time_stamp = m_scan_info.m_latest_time_stamp;
			scan_open_pkt.m_oldest_time_stamp = m_scan_info.m_oldest_time_stamp;
			scan_open_pkt.m_ttl = m_scan_info.m_ttl;
			scan_open_pkt.m_version_number = m_scan_info.m_version_number;
			scan_open_pkt.m_options = m_scan_info.m_options;
			scan_open_pkt.m_num_rows = m_scan_info.m_num_row;
		}

		std::string get_cur_row()
		{
			if (m_returned_row_count > 0)
			{
				return m_row_result_ptr->m_row_name;
			}
			else
			{
				return m_scan_info.m_start_row;
			}
		}

		std::string get_scan_start_row()
		{
			return m_scan_info.m_start_row;
		}

		std::string get_scan_end_row()
		{
			return m_scan_info.m_end_row;
		}

		unsigned get_returned_row_count()
		{
			return m_returned_row_count;
		}

		void async_get_next_row(handler_type handler);

		void dump_stat(scan_stat &upper_stat)
		{
			upper_stat.m_local_process_time += m_scan_stat.m_local_process_time;
			upper_stat.m_scan_open_time += m_scan_stat.m_scan_open_time;
			upper_stat.m_scan_open_times += m_scan_stat.m_scan_open_times;
			upper_stat.m_scan_next_time += m_scan_stat.m_scan_next_time;
			upper_stat.m_scan_next_times += m_scan_stat.m_scan_next_times;
			upper_stat.m_scan_record_count += m_scan_stat.m_scan_record_count;

			m_scan_stat.reset();
		}

	private:
		void on_scan_open_response(uint64_t request_time, protocol::packet_ptr pkt_ptr, const error_code &ec);

		void on_scan_next_response(uint64_t request_time, protocol::packet_ptr pkt_ptr, const error_code &ec);

		void on_scan_close_response(protocol::packet_ptr pkt_ptr, const error_code &ec);

		void do_scan_open();

		void do_scan_next();

		void do_scan_close();

		void on_scan_open_fail(const error_code &ec);

		void on_scan_next_fail(const error_code &ec);

		void on_error(const error_code &ec);

		void dispatch(row_result_ptr row_ptr, const error_code &ec);
	protected:
		ptr_type this_ptr()
		{
			return ptr_type(this);
		}

	private:
		endpoint_type m_server_endpoint;
		session_manager &m_session_manager;
		const client_options &m_options;
		tablet_scan_info m_scan_info;
		scan_block m_scan_block;
		unsigned m_returned_row_count;

		handler_type m_request_handler;
		row_result::ptr_type m_row_result_ptr;

		scan_open_request_ptr m_scan_open_pkt_ptr;
		scan_next_request_ptr m_scan_next_pkt_ptr;
		scan_close_request_ptr m_scan_close_pkt_ptr;

		boost::shared_ptr<boost::asio::deadline_timer> m_timer;

		bool m_started;
		unsigned m_retry_times;

		uint64_t m_scanner_id;

		scan_stat m_scan_stat;
	};
	
	typedef tablet_scanner::ptr_type tablet_scanner_ptr;
}

#endif

